package io.github.kimmking.gateway.outbound.okhttp;
import io.github.kimmking.gateway.filter.HeaderHttpResponseFilter;
import io.github.kimmking.gateway.filter.HttpRequestFilter;
import io.github.kimmking.gateway.filter.HttpResponseFilter;
import io.github.kimmking.gateway.outbound.httpclient4.NamedThreadFactory;
import io.github.kimmking.gateway.router.HttpEndpointRouter;
import io.github.kimmking.gateway.router.RandomHttpEndpointRouter;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;

import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.reactor.IOReactorConfig;


import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class OkhttpOutboundHandler {
    private CloseableHttpAsyncClient httpclient;
    private ExecutorService proxyService;
    private List<String> backendUrls;
    HttpEndpointRouter router = new RandomHttpEndpointRouter();

    public OkhttpOutboundHandler(List<String> backends) {

        this.backendUrls = backends.stream().map(this::formatUrl).collect(Collectors.toList());

        int cores = Runtime.getRuntime().availableProcessors();
        long keepAliveTime = 1000;
        int queueSize = 2048;
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();//.DiscardPolicy();
        proxyService = new ThreadPoolExecutor(cores, cores,
                keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize),
                new NamedThreadFactory("proxyService"), handler);

    }

    private String formatUrl(String backend) {
        return backend.endsWith("/")?backend.substring(0,backend.length()-1):backend;
    }

    public void handle(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, HttpRequestFilter filter) {

        String backendUrl = router.route(this.backendUrls);
        proxyService.submit(()->handleResponse(backendUrl,fullRequest,ctx,filter));

    }
  private void handleResponse(String url,final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, HttpRequestFilter filter)
  {
      FullHttpResponse response = null;
      OkHttpClient client = new OkHttpClient();
      //构造request
      url = url + fullRequest.uri();
      Request request = new Request.Builder().url(url).build();
      filter.filter(fullRequest, ctx);

      try (Response resp = client.newCall(request).execute()) {
          byte[] body = resp.body().bytes();
          response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,Unpooled.wrappedBuffer(body));
          HttpResponseFilter respomseFilter = new HeaderHttpResponseFilter(url);
          respomseFilter.filter(response);
          ctx.write(response);
      }
      catch (Exception ex){
          response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.EXPECTATION_FAILED ,Unpooled.wrappedBuffer(ex.getMessage().getBytes()));
          ctx.write(response);
      }
      ctx.flush();
      ctx.close();
  }
}
